/*
 * Copyright (c) 2023. Shanghai HEADING information Engineering Co., Ltd. All rights reserved.
 */

package com.lovely602.boot.bootdemo.controller.java;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.lovely602.boot.bootdemo.domain.common.RestResponse;
import com.lovely602.boot.bootdemo.domain.common.log.Watcher;

import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lizhixing
 */
@Api(tags = "currentHashMap服务")
@RestController
@RequestMapping(value = "/v1/bff/web/currentHashMap", produces = "application/json;charset=utf-8")
@Slf4j
public class CurrentHashMapController {

  // 循环次数
  private static int LOOP_COUNT = 10000000;
  // 线程数量
  private static int THREAD_COUNT = 10;
  // 元素数量
  private static int ITEM_COUNT = 10;

  @GetMapping("wrong")
  public RestResponse<String> wrong() throws InterruptedException {
    ConcurrentHashMap<String, Long> concurrentHashMap = getData(ITEM_COUNT - 900);
    // 初始900个元素
    log.info("init size:{}", concurrentHashMap.size());

    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    // 使用线程池并发处理逻辑
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
      // 查询还需要补充多少元素
      int gap = ITEM_COUNT - concurrentHashMap.size();
      log.info("gap size:{}", gap);
      // 补充元素
      concurrentHashMap.putAll(getData(gap));
    }));
    // 等待所有任务完成
    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
    // 最后元素个数会是1000吗？
    log.info("finish size:{}", concurrentHashMap.size());
    return RestResponse.success("ok");
  }

  @GetMapping("right")
  public RestResponse<String> right() throws InterruptedException {
    ConcurrentHashMap<String, Long> concurrentHashMap = getData(ITEM_COUNT - 900);
    // 初始900个元素
    log.info("init size:{}", concurrentHashMap.size());

    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    // 使用线程池并发处理逻辑
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
      synchronized (concurrentHashMap) {
        // 查询还需要补充多少元素
        int gap = ITEM_COUNT - concurrentHashMap.size();
        log.info("gap size:{}", gap);
        // 补充元素
        concurrentHashMap.putAll(getData(gap));
      }
    }));
    // 等待所有任务完成
    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
    // 最后元素个数会是1000吗？
    log.info("finish size:{}", concurrentHashMap.size());
    return RestResponse.success("ok");
  }

  @GetMapping("normalUse")
  public RestResponse<Map<String, Long>> normalUse() throws InterruptedException {
    ConcurrentHashMap<String, Long> freqs = new ConcurrentHashMap<>(ITEM_COUNT);
    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, LOOP_COUNT).parallel().forEach(i -> {
      // 获得一个随机的key
      String key = "item" + ThreadLocalRandom.current().nextInt(ITEM_COUNT);
      synchronized (freqs) {
        if (freqs.containsKey(key)) {
          // key存在则+1
          freqs.put(key, freqs.get(key) + 1);
        } else {
          // key不存在则初始化为1
          freqs.put(key, 1L);
        }
      }
    }));
    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
    return RestResponse.success(freqs);
  }

  @GetMapping("goodUse")
  public RestResponse<Map<String, Long>> goodUse() throws InterruptedException {
    ConcurrentHashMap<String, LongAdder> freqs = new ConcurrentHashMap<>(ITEM_COUNT);
    ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
    forkJoinPool.execute(() -> IntStream.rangeClosed(1, LOOP_COUNT).parallel().forEach(i -> {
      String key = "item" + ThreadLocalRandom.current().nextInt(ITEM_COUNT);
      // 利用computeIfAbsent()方法来实例化LongAdder，然后利用LongAdder来进行线程安全计数
      freqs.computeIfAbsent(key, k -> new LongAdder()).increment();
    }));
    forkJoinPool.shutdown();
    forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
    // 因为我们的value是LongAdder而不是Long，所以需要做一次转换才能返回。
    Map<String, Long> result = freqs.entrySet().stream()
        .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().longValue()));
    return RestResponse.success(result);
  }

  @GetMapping("good")
  public RestResponse<Void> good() throws InterruptedException {

    Watcher.watch("normalUse");
    Map<String, Long> normalUse = normalUse().getData();
    // 校验元素数量
    Assert.isTrue(normalUse.size() == ITEM_COUNT, "normalUse size error");
    // 校验累计总数
    Assert.isTrue(
        normalUse.values().stream().mapToLong(aLong -> aLong).reduce(0, Long::sum) == LOOP_COUNT,
        "normalUse count error");
    Watcher.watch("goodUse");
    Map<String, Long> goodUse = goodUse().getData();
    Assert.isTrue(goodUse.size() == ITEM_COUNT, "goodUse size error");
    Assert.isTrue(
        goodUse.values().stream().mapToLong(aLong -> aLong).reduce(0, Long::sum) == LOOP_COUNT,
        "goodUse count error");
    return RestResponse.success();
  }

  // 帮助方法，用来获得一个指定元素数量模拟数据的ConcurrentHashMap
  private ConcurrentHashMap<String, Long> getData(int count) {
    return LongStream.rangeClosed(1, count).boxed()
        .collect(Collectors.toConcurrentMap(i -> UUID.randomUUID().toString(), Function.identity(),
            (o1, o2) -> o1, ConcurrentHashMap::new));
  }

}
